package com.aotain.project.apollo.topo;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.ApolloConfig_Oracle;
import com.aotain.common.ZkKafkaConfig;
import com.aotain.kafka.KafkaConfigUtil;
import com.aotain.kafka.KafkaProperties;
import com.aotain.project.apollo.bolt.AbnReverseStatisBolt;
import com.aotain.project.apollo.bolt.AbnSourceStatisBolt;
import com.aotain.project.apollo.bolt.AbnStatisBolt;
import com.aotain.project.apollo.bolt.AbnormalIntoHbaseBolt;
import com.aotain.project.apollo.bolt.BaseStatSumBoltDelay;
import com.aotain.project.apollo.bolt.IDCStatisBolt;
import com.aotain.project.apollo.bolt.PortAbnStatisBolt;
import com.aotain.project.apollo.bolt.PortDetectBolt;
import com.aotain.project.apollo.bolt.SQLInjectionBolt;
import com.aotain.project.apollo.bolt.SdsSessStat5MinBolt;
import com.aotain.project.apollo.bolt.SessSourBolt;
import com.aotain.project.apollo.bolt.UDIIAbnormalCalBolt;
import com.aotain.project.apollo.bolt.UDIIAbnormalCalReverseBolt;
import com.aotain.project.apollo.bolt.UDIIBaseStatSumBolt;
import com.aotain.project.apollo.bolt.UDIIBaseStatSumSourceBolt;
import com.aotain.project.apollo.bolt.UDIISourceFmtBolt;
import com.aotain.project.apollo.boltv2.ACKFloodBolt;
import com.aotain.project.apollo.boltv2.CCDetectBoltV2;
import com.aotain.project.apollo.boltv2.DNSAttackBolt;
import com.aotain.project.apollo.boltv2.ICMPFloodBolt;
import com.aotain.project.apollo.boltv2.SYNFloodBolt;
import com.aotain.project.apollo.boltv2.UDIIBotsBolt;
import com.aotain.project.apollo.boltv2.UDIIScanAttackBolt;
import com.aotain.project.apollo.boltv2.UDPFloodBolt;
import com.aotain.project.apollo.boltv2.ViolentCrackBolt;
import com.aotain.project.apollo.boltv2.XSSDetectBolt;



public class UDIIAbnormalLogCheckTopo {
 
	static int workersNum = 6;
	static int spoutNum = 12;
	static int boltNum = 6;
	static int sourcefmtBoltNum = 6;
	public static void main(String[] args) {  
   
		if (args.length != 5){
			   System.err.printf("Usage:<TopoName> <Workers> <SpoutThreadNum> <BoltThreadNum> <SourceFmtboltThreadNum>");
			   System.exit(1);
			}  
		
		workersNum = Integer.parseInt(args[1]);
		spoutNum = Integer.parseInt(args[2]);
		boltNum = Integer.parseInt(args[3]);
		sourcefmtBoltNum = Integer.parseInt(args[4]);
		
		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");

		Logger.getLogger(UDIIAbnormalLogCheckTopo.class).info("Version 20160303");
		
		ApolloConfig ap = new ApolloConfig();
		ApolloConfig_Oracle ap_ = new ApolloConfig_Oracle();
		
		String dbJson = ap.getDBObjectJson();
		String dbJson_ = ap_.getDBObjectJson();
		Logger.getLogger(UDIIAbnormalLogCheckTopo.class).info(dbJson);
		
		TopologyBuilder builder = new TopologyBuilder();
		
		builder.setSpout("UDIIsourceSpout", new KafkaSpout(new KafkaConfigUtil(z_kConfig.getZkConnServer(),KafkaProperties.UDII_Abnormal_Log_topic,KafkaProperties.zkRoot_UDII,KafkaProperties.Log_UDII_Abnormal_groupId ).getConfig()),spoutNum);
		builder.setBolt("UDIISourceFmtBolt", new UDIISourceFmtBolt(),sourcefmtBoltNum).shuffleGrouping("UDIIsourceSpout");
		builder.setBolt("UDIIBaseStatSumBolt", new UDIIBaseStatSumBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		
		builder.setBolt("UDIIBaseStatSumSourceBolt", new UDIIBaseStatSumSourceBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","SipStreamId", new Fields("sip"));
		
		builder.setBolt("UDIIAbnormalCalBolt", new UDIIAbnormalCalBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId",new Fields("dip"));
		builder.setBolt("UDIIAbnormalCalReverseBolt", new UDIIAbnormalCalReverseBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","SipStreamId",new Fields("dip"));
		builder.setBolt("PortDetectBolt", new PortDetectBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("CCDetectBoltV2", new CCDetectBoltV2(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("XSSDetectBolt", new XSSDetectBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("ICMPFloodBolt", new ICMPFloodBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("ViolentCrackBolt", new ViolentCrackBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		
		builder.setBolt("ACKFloodBolt", new ACKFloodBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("SYNFloodBolt", new SYNFloodBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("DNSAttackBolt", new DNSAttackBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("UDPFloodBolt", new UDPFloodBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));//用于替代v1版本中的DosUpPacketAbnorBolt
		builder.setBolt("SQLInjectionBolt", new SQLInjectionBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("UDIIScanAttackBolt", new UDIIScanAttackBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		builder.setBolt("UDIIBotsBolt", new UDIIBotsBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		
		builder.setBolt("AbnStatisBolt", new AbnStatisBolt(),boltNum)
			.fieldsGrouping("UDIIAbnormalCalBolt", new Fields("dip"))
			.fieldsGrouping("UDIIAbnormalCalReverseBolt",new Fields("sip"))
			//由于端口异常需要上报的是次数，这里将流量统一向下输出STAT
			.fieldsGrouping("PortDetectBolt", "STAT", new Fields("dip"))
			.fieldsGrouping("CCDetectBoltV2", new Fields("dip"))
			.fieldsGrouping("SQLInjectionBolt", new Fields("dip"))
			.fieldsGrouping("ACKFloodBolt",new Fields("dip"))
			.fieldsGrouping("SYNFloodBolt",new Fields("dip"))
			.fieldsGrouping("XSSDetectBolt",new Fields("dip"))
			.fieldsGrouping("DNSAttackBolt",new Fields("dip"))		
			.fieldsGrouping("UDPFloodBolt",new Fields("dip"))
			.fieldsGrouping("ICMPFloodBolt",new Fields("dip"))
			.fieldsGrouping("UDIIScanAttackBolt",new Fields("dip"))
			.fieldsGrouping("UDIIBotsBolt",new Fields("dip"))
			.fieldsGrouping("ViolentCrackBolt",new Fields("dip"));
		
		builder.setBolt("PortAbnStatisBolt", new PortAbnStatisBolt(),boltNum)
			.fieldsGrouping("PortDetectBolt", new Fields("dip"));
		
		builder.setBolt("BaseStatSumBoltDelay", new BaseStatSumBoltDelay(),boltNum)
			.fieldsGrouping("UDIIBaseStatSumBolt","DELAY", new Fields("IP"))
			.fieldsGrouping("AbnStatisBolt","DELAY", new Fields("IP"))
			.fieldsGrouping("UDIIBaseStatSumSourceBolt","DELAY", new Fields("IP"));
		//.fieldsGrouping("BaseStatSumSourceBolt","DELAY", new Fields("IP"));
		

		//---Modified by turk 2017-06-27 IDC相关功能不需要
//		builder.setBolt("IDCStatisBolt", new IDCStatisBolt(),1)
//		.fieldsGrouping("AbnStatisBolt", new Fields("IDC"))
//		.fieldsGrouping("UDIIBaseStatSumBolt", new Fields("IDC"));
		       
		builder.setBolt("SessSourBolt", new SessSourBolt(),boltNum)
			.fieldsGrouping("AbnStatisBolt", "DELAY_", new Fields("DIP"))
			.fieldsGrouping("PortDetectBolt", "DELAY_", new Fields("DIP"));
		
		builder.setBolt("AbnormalIntoHbaseBolt", new AbnormalIntoHbaseBolt(),boltNum)
			.shuffleGrouping("UDIIAbnormalCalBolt", "ABNORMAL_LOG")
			.shuffleGrouping("UDIIAbnormalCalReverseBolt", "ABNORMAL_LOG");
		
		builder.setBolt("SdsSessStat5MinBolt",new SdsSessStat5MinBolt(),boltNum)
			.fieldsGrouping("AbnStatisBolt", "DELAY", new Fields("IP"));
		
		builder.setBolt("AbnSourceStatisBolt", new AbnSourceStatisBolt(),boltNum)
			.fieldsGrouping("UDIIAbnormalCalReverseBolt", "SoureAbnArea", new Fields("sip"));
		
		builder.setBolt("AbnReverseStatisBolt", new AbnReverseStatisBolt(),boltNum)
			.fieldsGrouping("UDIIAbnormalCalReverseBolt", new Fields("sip"));
				
		
		Config conf = new Config();    
		conf.setNumWorkers(workersNum); 
		conf.put("mushroomserver", z_kConfig.getMushroomServer());
		//conf.put("brokerlist", z_kConfig.getBrokerList());
		conf.put("dbjson", dbJson);
		conf.put("dbjson_", dbJson_);
//		System.out.println(dbJson);
		conf.put("sqlinjection", z_kConfig.getSQLInjection());
		Logger.getLogger(UDIIAbnormalLogCheckTopo.class).info(dbJson);
		conf.put("zkConn", z_kConfig.getZkConnServer());

		conf.setNumAckers(0);
//		conf.put("storm.messaging.netty.max_wait_ms", 10000);
		conf.setDebug(false);
		
		if (args.length > 0) {  
			try {
					StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
				
			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			}	catch (AuthorizationException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}else {
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("AbnormalLogTopo", conf, builder.createTopology());
		}
	}

}
